package com.scala.Class

import java.util
import java.util.Collections

import com.java.config.{DistanceConfig, RedisConfig}
import com.java.util.{LocationUtils, SerializeUtil, TestRedisPool}
import com.scala.CaseClass.scoreInfo
import org.apache.spark.sql.{SQLContext, SparkSession}
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service

import scala.collection.mutable

/**
 * @Author: ZhangJin
 * @Date: 2020/9/23 19:37
 */
@Service(value = "NearStoreChance")
class NearStoreChance @Autowired()(val session: SparkSession) extends Serializable {

  // TODO:   获取数据
  def getSource(LocalName1: String, codelaterrentinfodata: String): util.Map[Array[Byte], Array[Byte]] = {
    // TODO:   加载hive数据进map集合  ------ LocalName 地方名
    val hc: SQLContext = session.sqlContext
    val LocalName: String = LocalName1.concat("附近的商店")
    import session.implicits._
    // TODO:   调用java函数判断redis中是否存在  hset(String key,String item,Object value)
    import scala.collection.JavaConverters._
    val Jedis = TestRedisPool.getJedisPoolInstance(RedisConfig.redisIp, Integer.parseInt(RedisConfig.redisPort),RedisConfig.redisPassword).getResource
    if (Jedis.exists(LocalName.getBytes)) {
      val bytesToBytes: util.Map[Array[Byte], Array[Byte]] = Jedis.hgetAll(LocalName.getBytes)
      bytesToBytes
    } else {
      try {
        // TODO: 没有就去hive中取  nameOfStore:String,Url:String,Location:String,consumerByAvg:String,rateStore:String,numberCommit:String,Type:String,LocationOfLatAndLong:String
        hc.sql("select * from copewithdata." + codelaterrentinfodata).as[scoreInfo].rdd.foreachPartition(
          eachPartition => {
            val value = TestRedisPool.getJedisPoolInstance(RedisConfig.redisIp, Integer.parseInt(RedisConfig.redisPort),RedisConfig.redisPassword).getResource
            eachPartition.foreach(each => {
              value.hset(LocalName.getBytes(), each.UUid.getBytes(), mutable.Map("UUid" -> each.UUid, "nameOfStore" -> each.nameOfStore, "Url" -> each.Url, "Location" -> each.Location, "consumerByAvg" -> each.consumerByAvg, "rateStore" -> each.rateStore, "numberCommit" -> each.numberCommit, "Type" -> each.Type, "LocationOfLatAndLong" -> each.LocationOfLatAndLong).asJava.toString.getBytes())
            }
            )
            // TODO:   释放资源
            TestRedisPool.returnResource(value)
          }
        )
        val bytesToBytes: util.Map[Array[Byte], Array[Byte]] = Jedis.hgetAll(LocalName.getBytes())
        // TODO:   释放资源
        TestRedisPool.returnResource(Jedis)
        bytesToBytes
      } catch {
        case ex: Exception => {
          println("No find table ...")
          null
        }
      }
    }
  }

  def calnearByStore(LatAndLong: String,Data:util.Map[Array[Byte], Array[Byte]],Key: String): Unit = {
    if (Data == null) {
      println("无数据")
    }
    import scala.collection.JavaConverters._
    /*  map的value变为list   想把每个value的key变为每个list中的其中一员 */
    val list = new util.ArrayList[Array[Byte]](Data.values())
    var DataListALL = Collections.synchronizedList(new util.ArrayList[util.Map[String,String]]())
    val value1 = TestRedisPool.getJedisPoolInstance(RedisConfig.redisIp, Integer.parseInt(RedisConfig.redisPort),RedisConfig.redisPassword).getResource
    session.sparkContext.parallelize(list.asScala,20)

      .foreachPartition(
        {
          eachPartition=>{
            val Jedis1 = TestRedisPool.getJedisPoolInstance(RedisConfig.redisIp, Integer.parseInt(RedisConfig.redisPort),RedisConfig.redisPassword).getResource
            eachPartition.foreach(each=>{
              try {
                var stringToString: util.Map[String, String] = SerializeUtil.mapStringToMap(new String(each,"UTF-8"))
                val d: java.lang.Double = JudgeIsNotRangeLatAndLong(stringToString.get("LocationOfLatAndLong"), LatAndLong)
                // TODO:   筛选数据
                if(d <= java.lang.Double.parseDouble(DistanceConfig.distance))
                  {
                    stringToString.put("distance",d.toString+"km")
                    Jedis1.hset(Key.getBytes(), stringToString.get("UUid").getBytes(),stringToString.toString.getBytes())
                  }
              }
               catch {
                case ex: Exception => {
                  ex.printStackTrace()
                  System.err.println(".....")
                }
              }
            }
            )
            TestRedisPool.returnResource(Jedis1)
          }
        }
      )
    value1.expire(Key.getBytes(),1000)
    TestRedisPool.returnResource(value1)
  }

  def  JudgeIsNotRangeLatAndLong(Data: String,BeforeData: String): java.lang.Double = {
    val strings: Array[String] = Data.split("  ")
    val strings1: Array[String] = BeforeData.split("  ")
    val i = 1000
    val distance = LocationUtils.getDistance(strings(1), strings(0), strings1(1), strings1(0)) / i
    println("the data of distance for objectpoint is " + distance + "公里")
    distance
  }

}




